package io.task.cc.network.spi.client;


import io.task.cc.network.ChannelContext;
import io.task.cc.network.Node;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.StandardSocketOptions;
import java.nio.channels.AsynchronousChannelGroup;
import java.nio.channels.AsynchronousSocketChannel;
import java.util.concurrent.*;

@Slf4j
@Setter
@Getter
public class Client {


    private final AsynchronousChannelGroup asynchronousChannelGroup;

    private ClientConfig clientConfig;
    private final ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);

    public Client(final ClientConfig clientConfig) throws IOException {
        super();
        this.clientConfig = clientConfig;
        this.asynchronousChannelGroup = AsynchronousChannelGroup.withThreadPool(clientConfig.groupExecutor);

        //通过业务来进行判断是否连接已经断了，这里只实现重连功能。
        //如果是网络闪断，那么需要重连。如果是自我关闭，那么则不用重连。
        autoReConncect();
    }


    public void asynConnect(Node serverNode) throws Exception {
        asynConnect(serverNode, null);
    }


    public void asynConnect(Node serverNode, Integer timeout) throws Exception {
        asynConnect(serverNode, null, null, timeout);
    }


    public void asynConnect(Node serverNode, String bindIp, Integer bindPort, Integer timeout) throws Exception {
        connect(serverNode, bindIp, bindPort, null, timeout, false);
    }


    public ClientChannelContext connect(Node serverNode) throws Exception {
        return connect(serverNode, null);
    }


    public ClientChannelContext connect(Node serverNode, Integer timeout) throws Exception {
        return connect(serverNode, null, 0, timeout);
    }

    /**
     * @param timeout 超时时间，单位秒
     */
    public ClientChannelContext connect(Node serverNode, String bindIp, Integer bindPort, ClientChannelContext initClientChannelContext, Integer timeout) throws Exception {
        return connect(serverNode, bindIp, bindPort, initClientChannelContext, timeout, true);
    }

    /**
     * @param timeout 超时时间，单位秒
     * @param isSyn   true: 同步, false: 异步
     */
    private ClientChannelContext connect(Node serverNode, String bindIp, Integer bindPort, ClientChannelContext initClientChannelContext, Integer timeout, boolean isSyn)
            throws Exception {


        boolean isReconnect = initClientChannelContext != null;


        long start = System.currentTimeMillis();
        AsynchronousSocketChannel asynchronousSocketChannel = AsynchronousSocketChannel.open(asynchronousChannelGroup);
        long end = System.currentTimeMillis();
        long iv = end - start;
        if (iv >= 100) {
            log.error("{}, open 耗时:{} ms", iv);
        }

        asynchronousSocketChannel.setOption(StandardSocketOptions.TCP_NODELAY, true);
        asynchronousSocketChannel.setOption(StandardSocketOptions.SO_REUSEADDR, true);
        asynchronousSocketChannel.setOption(StandardSocketOptions.SO_KEEPALIVE, true);

        InetSocketAddress bind = null;
        if (bindPort != null && bindPort > 0) {
            if (bindIp != null) {
                bind = new InetSocketAddress(bindIp, bindPort);
            } else {
                bind = new InetSocketAddress(bindPort);
            }
        }

        if (bind != null) {
            asynchronousSocketChannel.bind(bind);
        }
        ClientChannelContext channelContext = initClientChannelContext;
        //
        InetSocketAddress inetSocketAddress = new InetSocketAddress(serverNode.getIp(), serverNode.getPort());
        ConnectionCompletionVo attachment = new ConnectionCompletionVo(channelContext, this, isReconnect, asynchronousSocketChannel, serverNode, bindIp, bindPort);

        if (isSyn) {
            Integer realTimeout = timeout;
            if (realTimeout == null) {
                realTimeout = 5;
            }

            CountDownLatch countDownLatch = new CountDownLatch(1);
            attachment.setCountDownLatch(countDownLatch);
            try {
                asynchronousSocketChannel.connect(inetSocketAddress, attachment, clientConfig.getConnectionCompletionHandler());
            } catch (Throwable e) {
                countDownLatch.countDown();
                //
                clientConfig.getConnectionCompletionHandler().failed(e, attachment);
                return attachment.getClientChannelContext();
            }
            @SuppressWarnings("unused")
            boolean await = countDownLatch.await(realTimeout, TimeUnit.SECONDS);
            return attachment.getClientChannelContext();
        } else {
            try {
                asynchronousSocketChannel.connect(inetSocketAddress, attachment, clientConfig.getConnectionCompletionHandler());
            } catch (Throwable e) {
                clientConfig.getConnectionCompletionHandler().failed(e, attachment);
            }
            return null;
        }
    }

    /**
     * @param timeout 超时时间，单位秒
     */
    public ClientChannelContext connect(Node serverNode, String bindIp, Integer bindPort, Integer timeout) throws Exception {
        return connect(serverNode, bindIp, bindPort, null, timeout);
    }


    /**
     * @param timeout 单位秒
     */
    public void reconnect(ClientChannelContext channelContext, Integer timeout) throws Exception {
        connect(channelContext.getServerNode(), channelContext.getBindIp(), channelContext.getBindPort(), channelContext, timeout);
    }


    /**
     * 启动重连任务
     */
    private void autoReConncect() {
        final ReconnConf reconnConf = clientConfig.getReconnConf();
        if (reconnConf == null || reconnConf.getInterval() <= 0) {
            return;
        }

        final String id = clientConfig.getId();
        Runnable runnable = new Runnable() {
            @Override
            public void run() {
                while (!clientConfig.isStopped()) {
                    log.error("closeds:{}, connections:{}", clientConfig.closeds.size(), clientConfig.connections.size());
                    //log.info("准备重连");
                    LinkedBlockingQueue<ChannelContext> queue = reconnConf.getQueue();
                    ClientChannelContext channelContext = null;
                    try {
                        channelContext = (ClientChannelContext) queue.take();
                    } catch (InterruptedException e1) {
                        log.error(e1.toString(), e1);
                    }
                    if (channelContext == null) {
                        continue;
                        //						return;
                    }

                    if (channelContext.isRemoved) //已经删除的，不需要重新再连
                    {
                        continue;
                    }


                    long sleeptime = reconnConf.getInterval() - (System.currentTimeMillis() - channelContext.channelStat.timeInReconnQueue);
                    //log.info("sleeptime:{}, closetime:{}", sleeptime, timeInReconnQueue);
                    if (sleeptime > 0) {
                        try {
                            Thread.sleep(sleeptime);
                        } catch (InterruptedException e) {
                            log.error(e.toString(), e);
                        }
                    }

                    if (channelContext.isRemoved || !channelContext.isClosed) //已经删除的和已经连上的，不需要重新再连
                    {
                        continue;
                    } else {
                        ReconnQueueTask runnable = channelContext.getReconnRunnable();
                        if (runnable == null) {
                            synchronized (channelContext) {
                                runnable = channelContext.getReconnRunnable();
                                if (runnable == null) {
                                    runnable = new ReconnQueueTask(channelContext, Client.this, reconnConf.getThreadPoolExecutor());
                                    channelContext.setReconnRunnable(runnable);
                                }
                            }
                        }
                        runnable.execute();
                        //						reconnConf.getThreadPoolExecutor().execute(runnable);
                    }
                }
            }
        };
        scheduledExecutorService.schedule(runnable, 1, TimeUnit.SECONDS);

    }


    public boolean stop() {
        boolean ret = true;
        try {
            clientConfig.groupExecutor.shutdown();
        } catch (Exception e1) {
            log.error(e1.toString(), e1);
        }
        try {
            clientConfig.synThreadPoolExecutor.shutdown();
        } catch (Exception e1) {
            log.error(e1.toString(), e1);
        }

        clientConfig.setStopped(true);
        try {
            ret = ret && clientConfig.groupExecutor.awaitTermination(6000, TimeUnit.SECONDS);
            ret = ret && clientConfig.synThreadPoolExecutor.awaitTermination(6000, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            log.error(e.getLocalizedMessage(), e);
        }
        log.info("client resource has released");
        return ret;
    }
}
